package com.wzk.canalsync.queue;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.wzk.canalsync.canal.CanalClient;
import com.wzk.canalsync.utils.LogToFile;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.yetus.audience.InterfaceAudience;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

/**
 * @author wangzikang
 */

@Slf4j
@Component
@RequiredArgsConstructor
public class QueueHandler {

    @Value("${es.indexName}")
    private String indexName;

    public static final Logger logger = Logger.getLogger(QueueHandler.class.getName());

    private ReentrantLock lock = new ReentrantLock();

    private final ObjectMapper objectMapper;

    private final RestHighLevelClient esClient;

    //队列最大长度
    private final int queue_max_size = 100000;

    //操作队列存放要执行的操作节点
    private final ConcurrentLinkedQueue<OperateNode> queue = new ConcurrentLinkedQueue<OperateNode>();

    private final ExecutorService operatePool;


    @PreDestroy
    public void appShutdown() throws IOException {
        log.info("应用正在关闭..... 等待清理系统资源");
        try {
            //如果队列不是空的就将node转换成json存到txt文件
            if (!queue.isEmpty()){
                log.info("队列数据开始写入到txt文件.....  一共："+queue.size()+"条");
                //阻塞式获取
                lock.lock();
                exportQueueToFile("node_data");
                log.info("写入完毕!");
            }
        }catch (Exception e){
            log.error(e.getMessage());
        }finally {
            if (lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }

    //40s执行一次
    private final long taskTime = 40000;

    //定时任务批量消费 40s执行一次
    @Scheduled(fixedRate = taskTime)
    public void sendBatchRequest(){
        try {
            if (!queue.isEmpty() && lock.tryLock()){
                //开始时间
                long startTime = System.currentTimeMillis();
                log.info("获取到锁开始构建批量操作，当前队列大小：{}",queue.size());
                LogToFile.logToFile("1. -----------获取到锁开始批量操作,queueSize："+queue.size()+",定时任务["+(taskTime/1000)+"s]执行一次-----------",logger,"batch_request");
                //计算当前批量大小
                int batchSize = Math.min(queue.size(), 1000);
                LogToFile.logToFile("2. 当前批量大小："+batchSize,logger,"batch_request");
                //构建批量请求
                structBatchRequest(batchSize);
                long endTime = System.currentTimeMillis();
                LogToFile.logToFile("3. 当前批量请求耗时："+(endTime-startTime)+"ms",logger,"batch_request");
                LogToFile.logToFile("4. ----------------------------批量请求结束queueSize："+queue.size()+"---------------------------",logger,"batch_request");
                LogToFile.logToFile("执行完毕",logger,"batch_request");
            }
        }catch (Exception e){
            log.error("定时任务执行出现异常：{}",e.getMessage());
            LogToFile.logToFile("定时任务执行出现异常:"+e.getMessage(),logger,"timer_task_err");
        }finally {
            if (lock.isHeldByCurrentThread()){
                lock.unlock();
            }
        }
    }

    //插入数据到队列
    public void push(OperateNode node) {
        try {
            //判断队列长度是否达到最大
            if (queue.size() >= queue_max_size) {
                log.info("队列满了！开始写入日志");
                LogToFile.logToFile(node.toString(),logger,"queue_full");
                return;
            }
            //队列没满
            queue.add(node);
        }catch (Exception e){
            log.error("队列插入出现异常:{}",e.getMessage());
            LogToFile.logToFile("队列插入出现异常:"+e.getMessage(),logger,"queue_push_err");
        }
    }

    //构造批量请求
    public void structBatchRequest(int batchSize) {
        try {
            if (!queue.isEmpty()){
                int currCount = batchSize;
                BulkRequest bulkRequest = new BulkRequest();
                while (currCount > 0 && !queue.isEmpty()) {
                    OperateNode node = queue.poll();
                    if (node == null) {
                        break;
                    }
                    LogToFile.logToFile(node.toString(),logger,"queue_nodes");
                    if ("insert".equalsIgnoreCase(node.getOperate())){
                        IndexRequest indexRequest = new IndexRequest();
                        indexRequest.index(indexName);
                        indexRequest.id(node.getParamsMap().get("id").toString());
                        indexRequest.source(objectMapper.writeValueAsString(node.getParamsMap()), XContentType.JSON);
                        bulkRequest.add(indexRequest);
                    }else if ("update".equalsIgnoreCase(node.getOperate())){
                        UpdateRequest updateRequest = new UpdateRequest();
                        updateRequest.index(indexName);
                        updateRequest.id(node.getParamsMap().get("id").toString());
                        updateRequest.doc(objectMapper.writeValueAsString(node.getParamsMap()), XContentType.JSON);
                        bulkRequest.add(updateRequest);
                    }else if ("delete".equalsIgnoreCase(node.getOperate())){
                        DeleteRequest deleteRequest = new DeleteRequest().index(indexName);
                        deleteRequest.id(node.getParamsMap().get("id").toString());
                        bulkRequest.add(deleteRequest);
                    }
                    currCount--;
                }
                //发送批量请求
                operatePool.submit(() -> {
                    try {
                        esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    }catch (Exception e){
                        if (e.getMessage().contains("200")){
                            log.info("批量请求发送成功！");
                        }else {
                            log.error("批量请求执行失败！出现：{}",e.getMessage());
                            LogToFile.logToFile("es批量请求出现异常！exceptionMessage:【"+e.getMessage()+"】",logger,"es_bulk_err");
                        }
                    }
                });
            }
        } catch (Exception e) {
            log.error("构建批量请求出现异常：{}",e.getMessage());
            LogToFile.logToFile("构建批量请求出现异常："+e.getMessage(),logger,"batch_request_err");
        }
    }

    //如果队列满了写入到文件中
    public void exportQueueToFile(String fileName) throws IOException {
        try {
            String queueDataFileName = "queue/"+ fileName + ".txt";
            Path dataFilePath = Paths.get(queueDataFileName);
            // 创建目录，如果目录不存在
            Files.createDirectories(dataFilePath.getParent());
            // 如果文件不存在，创建新文件；如果文件已存在，以追加方式打开
            // 使用 BufferedWriter 写入文件
            try (BufferedWriter writer = Files.newBufferedWriter(dataFilePath, StandardOpenOption.CREATE, StandardOpenOption.APPEND)) {
                while (!queue.isEmpty()) {
                    OperateNode operateNode = queue.poll();
                    String nodeJson = objectMapper.writeValueAsString(operateNode);  // 转为JSON字符串
                    writer.write(nodeJson);  // 写入数据
                    writer.newLine();  // 换行
                }
            }
        }catch (Exception e){
            log.error("queue数据写入txt出现异常！异常：{}",e.getMessage());
        }
    }

}
